package com.changdu.seatunnel.admin.controller;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.DeleteMapping;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.PutMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import com.changdu.seatunnel.admin.entity.SyncConfig;
import com.changdu.seatunnel.admin.pojo.CopySyncConfigGenerate;
import com.changdu.seatunnel.admin.pojo.KafkaConfig;
import com.changdu.seatunnel.admin.pojo.SourceConfig;
import com.changdu.seatunnel.admin.pojo.SyncConfigGenerate;
import com.changdu.seatunnel.admin.pojo.TargetConfig;
import com.changdu.seatunnel.admin.service.SyncConfigService;
import com.changdu.seatunnel.admin.util.SyncConfigUtils;

@RestController
@RequestMapping("/api/sync/config")
public class SyncConfigController {
    
    @Autowired
    private SyncConfigService syncConfigService;
    
    @GetMapping("/{id}")
    public SyncConfigGenerate getById(@PathVariable Long id) {
        SyncConfig config = syncConfigService.getById(id);
        return SyncConfigUtils.toSyncConfigGenerate(config);
    }
    
    @PostMapping
    public boolean save(@RequestBody SyncConfig config) {
        return syncConfigService.save(config);
    }
    
    @PutMapping
    public boolean update(@RequestBody SyncConfig config) {
        return syncConfigService.updateById(config);
    }
    
    @DeleteMapping("/{id}")
    public boolean delete(@PathVariable Long id) {
        return syncConfigService.removeById(id);
    }
    
    @PostMapping("/generate")
    public void generateConfig(@RequestBody SyncConfigGenerate configData) {
    	SyncConfig config = SyncConfigUtils.build(configData);
    	// 先保存配置
    	syncConfigService.save(config);
    	// 基于模板生成任务配置
    	syncConfigService.generateConfigWithId(config.getId(), configData);
    }

    @PostMapping("/generate/{id}")
    public void updateGenerateConfig(@PathVariable Long id, @RequestBody SyncConfigGenerate configData) {
    	SyncConfig config = SyncConfigUtils.buildById(id, configData);
    	// 先更新配置
    	syncConfigService.updateById(config);
    	// 基于模板生成任务配置
    	syncConfigService.updateGenerateConfigWithId(config.getId(), configData);
    }
    
    /**
     * 按数据中心拷贝并生成配置
     * @param id
     */
    @PostMapping("/generate/copy/{id}")
    public void copyGenerateConfig(@PathVariable Long id, @RequestBody CopySyncConfigGenerate configData) {
    	SyncConfigGenerate generate = getById(id);
    	SourceConfig sourceConfig = generate.getSourceConfig();
    	sourceConfig.setDataCenter(configData.getSourceDataCenter());
    	sourceConfig.setInstance(configData.getSourceInstance());
    	
    	TargetConfig targetConfig = generate.getTargetConfig();
    	targetConfig.setDataCenter(configData.getTargetDataCenter());
    	targetConfig.setInstance(configData.getTargetInstance());
    	generate.setId(null);
    	
    	// 绑定 topic
    	generate.setKafkaConfig(configData.getKafkaConfig());
    	
    	// 基于规则替换属性
    	SyncConfigGenerate newConfig = SyncConfigUtils.replaceProperties(generate, 
    			configData.getSourceDbReplaceRegex(), configData.getSourceTbReplaceRegex(),
    			configData.getTargetDbReplaceRegex(), configData.getTargetTbReplaceRegex(), configData.getDefaultValues());
    	generateConfig(newConfig);
    }
    
    @PostMapping("/generate/kafkaTopicCheck")
    public boolean kafkaTopicCheck(@RequestBody KafkaConfig kafkaConfig) {
    	return syncConfigService.countKafkaWithConsumer(kafkaConfig.getKafkaTopic(), kafkaConfig.getKafkaConsumerGroup()) == 0;
    } 
    
}